package com.example.rocketmq.demo.producer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * 过滤消息生产者
 */
@Slf4j
@Component
public class FilterProducer {

    @Autowired
    @Qualifier("defaultMQProducer")
    private DefaultMQProducer producer;

    /**
     * 发送带标签过滤的消息
     *
     * @param content 消息内容
     * @param tag 消息标签
     * @return 发送结果
     */
    public SendResult sendWithTag(String content, String tag) throws Exception {
        Message message = new Message(
                RocketMQConstant.TOPIC_FILTER,
                tag,
                content.getBytes(StandardCharsets.UTF_8)
        );

        SendResult sendResult = producer.send(message);
        log.info("带标签过滤的消息发送结果: {}, tag: {}", sendResult, tag);
        return sendResult;
    }

    /**
     * 发送带属性过滤的消息
     *
     * @param content 消息内容
     * @param tag 消息标签
     * @param key 属性键
     * @param value 属性值
     * @return 发送结果
     */
    public SendResult sendWithProperty(String content, String tag, String key, String value) throws Exception {
        Message message = new Message(
                RocketMQConstant.TOPIC_FILTER,
                tag,
                content.getBytes(StandardCharsets.UTF_8)
        );

        // 设置消息属性，用于SQL过滤
        message.putUserProperty(key, value);

        SendResult sendResult = producer.send(message);
        log.info("带属性过滤的消息发送结果: {}, tag: {}, property: {}={}", sendResult, tag, key, value);
        return sendResult;
    }
}
